package com.niodata.dt.fs;

import java.io.File;
import java.io.FileInputStream;
import java.io.FileNotFoundException;
import java.io.FileOutputStream;
import java.io.IOException;
import java.net.URI;
import java.nio.file.AccessDeniedException;
import java.nio.file.Files;
import java.nio.file.NoSuchFileException;
import java.util.ArrayList;
import java.util.List;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FSDataInputStream;
import org.apache.hadoop.fs.FSDataOutputStream;
import org.apache.hadoop.fs.FileStatus;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.fs.permission.FsPermission;
import org.apache.hadoop.util.Progressable;

public class DpLocalFileSystem extends FileSystem {

  public static final String FS_DPLOCAL_ROOT_DIR = "source.path.root";
  private URI uri;
  private String rootPath;
  private Path workingDir;

  @Override
  public void initialize(URI name, Configuration conf) throws IOException {
    super.setConf(conf);
    super.initialize(name, conf);
    this.rootPath = conf.get(FS_DPLOCAL_ROOT_DIR);
    if (!this.rootPath.endsWith("/")) {
      this.rootPath = this.rootPath + "/";
    }
    File file = new File(this.rootPath);
    if (!file.exists()) {
      throw new FileNotFoundException(file.getAbsolutePath());
    }
    if (file.isFile()) {
      throw new IOException("root path is file:" + file.getAbsolutePath());
    }
    String url = "dpLocal://" + this.rootPath;
    uri = URI.create(url);
    workingDir = new Path(System.getProperty("user.dir"));
  }


  @Override
  public URI getUri() {
    return this.uri;
  }

  @Override
  public FSDataInputStream open(Path f, int bufferSize) throws IOException {
    this.checkRootPath(f);
    File file = new File(f.toString());
    return new FSDataInputStream(new LocalFsInputStream(file, new FileInputStream(file)));
  }

  @Override
  public FSDataOutputStream create(Path f, FsPermission permission, boolean overwrite, int bufferSize,
                                   short replication, long blockSize, Progressable progress)
      throws IOException {
    this.checkRootPath(f);
    return new FSDataOutputStream(new FileOutputStream(Path.getPathWithoutSchemeAndAuthority(f).toString()), null);
  }

  @Override
  public FSDataOutputStream append(Path f, int bufferSize, Progressable progress) throws IOException {
    this.checkRootPath(f);
    return new FSDataOutputStream(new FileOutputStream(Path.getPathWithoutSchemeAndAuthority(f).toString(), true), null);
  }

  @Override
  public boolean rename(Path src, Path dst) throws IOException {
    this.checkRootPath(src);
    this.checkRootPath(dst);
    File source = new File(Path.getPathWithoutSchemeAndAuthority(src).toString());
    if (!source.exists()) {
      throw new IOException("source file not exists:" + src);
    }
    File target = new File(Path.getPathWithoutSchemeAndAuthority(dst).toUri());
    if (target.exists()) {
      throw new IOException("target file exists:" + dst);
    }

    return source.renameTo(target);
  }

  @Override
  public boolean delete(Path f, boolean recursive) throws IOException {
    this.checkRootPath(f);
    File file = new File(Path.getPathWithoutSchemeAndAuthority(f).toString());
    if (!file.exists()) {
      return true;
    }
    FileUtil.deleteFileOrDir(file, recursive);
    return true;
  }

  @Override
  public FileStatus[] listStatus(Path f) throws IOException {
    this.checkRootPath(f);
    File file = new File(Path.getPathWithoutSchemeAndAuthority(f).toString());
    if (!file.exists()) {
      System.out.println("S" + file.getAbsolutePath()+"S");
      throw new FileNotFoundException(file.getAbsolutePath());
    }
    if (file.isFile()) {
      throw new IOException("file is not directory:" + file.getAbsolutePath());
    }
    File[] files = file.listFiles();
    if (files == null) {
      return new FileStatus[0];
    }
    List<FileStatus> statuses = new ArrayList<>(files.length);
    for (int i = 0; i < files.length; i++) {
      File tf = files[i];
      FileStatus status = null;
      if (Files.isSymbolicLink(tf.toPath())) {
        status = new FileStatus(0L, false, 1, FileUtil.SIZE_128MB, file.lastModified(), f);
        try {
        status.setSymlink(new Path(Files.readSymbolicLink(tf.toPath()).toString()));
        statuses.add(status);
        continue;
        } catch (Exception ex) {
          continue;
        }
      }
      if (tf.isFile()) {
        status = new FileStatus(tf.length(), false, 1, FileUtil.SIZE_128MB, tf.lastModified(),
          new Path(tf.getAbsolutePath()));
      } else {
        status = new FileStatus(tf.length(), true, 1, 0, tf.lastModified(), new Path(tf.getAbsolutePath()));
      }
      statuses.add(status);
    }
    return statuses.toArray(new FileStatus[statuses.size()]);
  }

  @Override
  public Path getWorkingDirectory() {
    return workingDir;
  }

  @Override
  public void setWorkingDirectory(Path newDir) {
    this.workingDir = newDir;
  }

  @Override
  public boolean mkdirs(Path f, FsPermission permission) throws IOException {
    checkRootPath(f);
    File file = new File(Path.getPathWithoutSchemeAndAuthority(f).toString());
    return file.mkdirs();
  }

  @Override
  public FileStatus getFileStatus(Path f) throws IOException {
    checkRootPath(f);
    File file = new File(Path.getPathWithoutSchemeAndAuthority(f).toString());
    if (!file.exists()) {
      throw new FileNotFoundException(Path.getPathWithoutSchemeAndAuthority(f).toString());
    }
    FileStatus status = null;
    if (Files.isSymbolicLink(file.toPath())) {
      status = new FileStatus(0, false, 1, FileUtil.SIZE_128MB, file.lastModified(), f);
      status.setSymlink(new Path(Files.readSymbolicLink(file.toPath()).toString()));
      return status;
    }
    if (file.isFile()) {
      status = new FileStatus(file.length(), false, 1, FileUtil.SIZE_128MB, file.lastModified(), f);
    } else if (file.isDirectory()){
      status = new FileStatus(0, true, 1, FileUtil.SIZE_128MB, file.lastModified(), f);
    }
    return status;
  }

  private void checkRootPath(Path f) throws IOException {
    String path = Path.getPathWithoutSchemeAndAuthority(f).toString();
    if (!path.endsWith("/")) {
      path = path + "/";
    }
    if (!path.startsWith(rootPath)) {
      throw new IOException("No permission on path:" + path);
    }
  }

  @Override
  public void copyToLocalFile(Path src, Path dst) throws IOException {
    super.copyToLocalFile(false, src, dst, true);
  }

  @Override
  public void copyToLocalFile(boolean delSrc, Path src, Path dst) throws IOException {
    super.copyToLocalFile(delSrc, src, dst, true);
  }
}
